package com.atguigu.pro

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Slide, Table}
import org.apache.flink.types.Row

/**
 * @description: xxx
 * @time: 2020/8/4 16:10
 * @author: baojinlong
 * */
object HotItemsWithSql {
  def main(args: Array[String]): Unit = {
    // 创建一个流处理执行环境
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    // 设置时间语义
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 设置全局并行度1
    environment.setParallelism(1)

    // 从文件中读取数据
    val inputStream: DataStream[String] = environment.readTextFile("C:/codes/scala/FlinkTutorial/src/main/resources/UserBehavior-short.csv")


    // 将数据转换成样例类类型,并且提取timestamp定义watermark
    val dataStream: DataStream[UserBehavior] = inputStream
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        UserBehavior(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000)

    // 定义表执行环境
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance
      .useBlinkPlanner
      .inStreamingMode
      .build

    val tableEnvironment: StreamTableEnvironment = StreamTableEnvironment.create(environment, settings)
    dataStream.print()
    // 基于DataStream创建table,timestamp定义成事件时间字段,方便后续sql调用,timestamp.rowtime即为事件时间
    val dataTable: Table = tableEnvironment.fromDataStream(dataStream, 'itemId, 'behavior, 'timestamp.rowtime as 'ts)
    // dataTable.toAppendStream[Row].print("xxx")
    // table api 进行开创聚合统计
    val aggTable: Table = dataTable
      .filter('behavior === "pv")
      // 指定当前字段
      .window(Slide over 1.hours every 5.minutes on 'ts as 'sw)
      .groupBy('itemId, 'sw)
      .select('itemId, 'sw.end as 'windowEnd, 'itemId.count as 'cnt)

    tableEnvironment.createTemporaryView("aggtable", aggTable, 'itemId, 'windowEnd, 'cnt)
    // 用sql实现topN的选取
    val resultTable: Table = tableEnvironment.sqlQuery(
      """
        |select * from
        |(select *,
        |ROW_NUMBER()
        |over (partition by windowEnd order by cnt desc)
        |as row_num
        |from aggtable
        |)
        |where row_num <=5
        |""".stripMargin
    )

    // 纯sql实现
    tableEnvironment.createTemporaryView("datatable", dataStream, 'itemId, 'behavior, 'timestamp.rowtime as 'ts)
    val tableSqlResult: Table = tableEnvironment.sqlQuery(
      """
        |select * from
        |(select *,
        |row_number()
        |over (partition by windowEnd order by cnt desc)
        |as row_num
        |from (
        | select
        |   itemId,
        |   hop_end(ts,interval '5' minute,interval '1' hour) as windowEnd,
        |   count(itemId) cnt
        | from datatable
        | where behavior = 'pv'
        | group by (
        |   itemId,
        |   hop(ts,interval '5' minute,interval '1' hour)
        | )
        |)
        |)
        |where row_num <=5
        |""".stripMargin
    )


    resultTable.toRetractStream[Row].print("resultTable")
    tableSqlResult.toRetractStream[Row].print("tableSqlResult")
    environment.execute("proj")
  }
}
